package org.bikelab.se3.mr;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.bikelab.se3.rdf.reader.RdfReader;
import org.bikelab.se3.rdf.reader.RdfReaderException;
import org.bikelab.se3.rdf.reader.RdfReaderFactory;
import org.semanticweb.yars.nx.Node;

public class NQuadContextIndexer extends Mapper<LongWritable, Text, Text, Text> {
	public void map(LongWritable key, Text value, Context context)
			throws IOException {
		try {
			RdfReader quadReader = RdfReaderFactory.getRdfReader(context.getConfiguration());
			quadReader.read(value.toString());
			while(quadReader.hasNext()) {
				Node[] quad = quadReader.next();
				String triple = quad[0].toN3() + " " + quad[1].toN3() + " " + quad[2].toN3() + " .";
				context.write(new Text(quad[3].toN3()), new Text(triple));
			}
		} catch (RdfReaderException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}
